package cn.pengpeng.curd;

import java.io.BufferedReader;
import java.io.FileReader;
import java.lang.reflect.Type;
import java.util.HashMap;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;

import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;

public class AppFavProcess {
	
	public static void main(String[] args) throws Exception {
		
		/**
		 * 解析原始日志数据的代码
		 */
		// 读当日的数据日志
		BufferedReader br = new BufferedReader(new FileReader("F:\\mrdata\\appfav\\2017-11-09\\app.log"));
		
		// 按行解析
		String line = null;
		HashMap<String, HashMap<String,Integer>> favMap = new HashMap<>();
		while((line=br.readLine())!=null) {
			String[] split = line.split(",");
			String day = split[0];
			String uid = split[1];
			String app = split[2];
			
			if(favMap.containsKey(uid)) {
				HashMap<String, Integer> appMap = favMap.get(uid);
				if(appMap.containsKey(app)) {
					appMap.put(app, appMap.get(app)+1);
				}else {
					appMap.put(app, 1);
				}
				
			}else {
				HashMap<String, Integer> appMap = new HashMap<>();
				appMap.put(app, 1);
				favMap.put(uid, appMap);
				
			}
		}
		br.close();
		
		/**
		 * 插入数据到hbase的代码
		 */
		
		Gson gson = new Gson();
		Type type = new TypeToken<HashMap<String,Integer>>() {}.getType();
		
		// 创建一个hbase连接
		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", "hdp27-01:2181,hdp27-02:2181,hdp27-03:2181");
		Connection conn = ConnectionFactory.createConnection(conf);
		Table table = conn.getTable(TableName.valueOf("t_app_fav"));
		
		
		// 更新当日偏好 
		Set<String> users = favMap.keySet();
		for (String uid : users) {
			HashMap<String, Integer> appMap = favMap.get(uid);
			String appJson = gson.toJson(appMap,type);
			Put put = new Put(uid.getBytes());
			put.addColumn("f1".getBytes(), uid.getBytes(), appJson.getBytes());
			table.put(put);
		}
		
		
		// 更新累计偏好
		Type type2 = new TypeToken<HashMap<String,Float>>() {}.getType();
		// 1、先取出累计值
		Scan scan = new Scan();
		scan.addFamily("f2".getBytes());
		ResultScanner scanner = table.getScanner(scan);
		
		boolean flag = true;
		
		for (Result result : scanner) {
			flag = false;
			
			String uid = new String(result.getRow());
			byte[] valueArr = result.getValue("f2".getBytes(), uid.getBytes());
			String json = new String(valueArr);  // 该用户的累计偏好信息json
			
			HashMap<String,Float> histAppFav = gson.fromJson(json, type2);
			HashMap<String, Integer> appFav = favMap.get(uid);
			
			
			// 2、比对当日值和累计值，并更新累计值
			HashMap<String,Float> combined  = combine(histAppFav,appFav);
			
			// 3、将更新之后的累计值put回hbase
			String combinedJson = gson.toJson(combined, type2);
			
			Put put = new Put(uid.getBytes());
			put.addColumn("f2".getBytes(), uid.getBytes(), combinedJson.getBytes());
			table.put(put);
			
		}
		
		// 如果这是第一次，则直接将当日偏好数据写入历史偏好
		if(flag) {
			for (String uid : users) {
				HashMap<String, Integer> appMap = favMap.get(uid);
				String appJson = gson.toJson(appMap,type);
				Put put = new Put(uid.getBytes());
				put.addColumn("f2".getBytes(), uid.getBytes(), appJson.getBytes());
				table.put(put);
			}
			
		}
		
		
		table.close();
		conn.close();
		
	}

	/**
	 * 合并当日偏好和历史累计偏好
	 * @param histAppFav
	 * @param appFav
	 * @return
	 */
	private static HashMap<String, Float> combine(HashMap<String, Float> histAppFav, HashMap<String, Integer> appFav) {

		// 当日出现的app，累计也出现；     当日出现的app，累计中没有；   当日没出现但在累计中出现的app；
		
		// 先遍历当日，已进行累加
		Set<Entry<String, Integer>> entrySet = appFav.entrySet();
		for (Entry<String, Integer> entry : entrySet) {
			// 如果在历史累计中也有这个app
			if(histAppFav.containsKey(entry.getKey())){
				// 将历史中的偏好值+当日该app的偏好值
				histAppFav.put(entry.getKey(), histAppFav.get(entry.getKey())+entry.getValue());
			}else {
				// 对历史数据新增一个app
				histAppFav.put(entry.getKey(), (float)entry.getValue());
			}
			
		}
		
		// 再遍历历史数据，以进行衰减
		
		Set<Entry<String, Float>> entrySet2 = histAppFav.entrySet();
		for (Entry<String, Float> entry : entrySet2) {
			// 如果历史数据中的app在当日中没有，则衰减
			if(!appFav.containsKey(entry.getKey())) {
				histAppFav.put(entry.getKey(), entry.getValue()*0.7f);
			}
		}
		
		return histAppFav;
	}
	

}
